#include "FIFOSafeQueue.h"
#include <atomic>
#include <cassert>
#include <cstddef>
#include <cstdint>
#include <cstdlib>
#include <cstring>
#include <memory>
#include <sstream>
#include <unistd.h>
#include "PlatLogger.h"
#include "Util.h"
#include "XnccEventfd.h"
#include "XnccType.h"
namespace xncc {
namespace msg {
    using namespace xncc::shared;
    using namespace xncc::types;
    class FIFOSafeElement
    {
      public:
        FIFOSafeElement(types::queue_size_t size) : mask(size - 1)
        {
            if (!is_power_of_2(size)) {
                throw std::logic_error("size must be power of 2");
            }
            in   = 0;
            out  = 0;
            data = reinterpret_cast<char*>(malloc(size));
            assert(data != nullptr);
            segmentData = reinterpret_cast<char*>(malloc(size));
            assert(segmentData != nullptr);
        }
        ~FIFOSafeElement()
        {
            free(data);
            free(segmentData);
        }

        [[nodiscard]] inline types::queue_size_t capacity() const { return mask + 1; }

        bool write(const char* buffer, types::queue_size_t len)
        {
            types::queue_size_t l;
            const auto          cached_out = out;
            const auto          cached_in  = in;
            l                              = capacity() - static_cast<types::queue_size_t>((cached_in - cached_out));
            if (len > l) {
                return false;
            }
            const types::queue_size_t off = cached_in & mask;
            l                             = std::min(len, static_cast<types::queue_size_t>(capacity() - off));
            memcpy(data + off, buffer, l);
            memcpy(data, buffer + l, len - l);
            return true;
        }

        void writeOver(types::queue_size_t len)
        {
            smp_wmb();
            const auto cached_in = in;
            const auto result    = static_cast<types::queue_size_t>(cached_in + len);
            in                   = result;
        }

        char* read(types::queue_size_t len)
        {
            types::queue_size_t l;
            const auto          cached_out = out;
            const auto          cached_in  = in;
            l                              = cached_in - cached_out;
            if (len > l) {
                return nullptr;
            }
            const types::queue_size_t off = cached_out & mask;
            l                             = std::min(len, static_cast<types::queue_size_t>(capacity() - off));
            if (likely(len == l)) {
                return data + off;
            }
            else {
                memcpy(segmentData, data + off, l);
                memcpy(segmentData + l, data, len - l);
                return segmentData;
            }
        }

        void readOver(types::queue_size_t len)
        {
            smp_wmb();
            const auto cached_out = out;
            const auto result     = static_cast<types::queue_size_t>(cached_out + len);
            out                   = result;
        }

      private:
        inline bool is_power_of_2(uint64_t n) { return (n != 0 && ((n & (n - 1)) == 0)); }

        inline uint64_t roundup_pow_of_two(uint64_t n)
        {
            n |= n >> 1;
            n |= n >> 2;
            n |= n >> 4;
            n |= n >> 8;
            n |= n >> 16;
            n |= n >> 32;
            n += 1;
            return n;
        }

        inline uint64_t rounddown_pow_of_two(uint64_t n) { return roundup_pow_of_two(n) >> 1; }

      private:
        const types::queue_size_t             mask;
        volatile __attribute__((aligned(64))) types::queue_size_t in;
        volatile __attribute__((aligned(64))) types::queue_size_t out;
        char*                                                     data;
        char*                                                     segmentData;
    };

    static std::atomic<types::queue_id_t> INIT_QUEUE_ID(1);
    FIFOSafeQueue::FIFOSafeQueue(queue_size_t size, bool autoExpand)
        : autoExpand_(autoExpand), lastReadIndex_(0), isNeedNotify_(false), queueId_(INIT_QUEUE_ID.fetch_add(1))
    {
        bufferList_[0] = std::make_shared<FIFOSafeElement>(size);
        readptr_       = bufferList_[0];
        writeptr_      = bufferList_[0];
    }
    void FIFOSafeQueue::setNeedNotify()
    {
        notifyfd_     = (std::make_shared<sys::EventFd>());
        isNeedNotify_ = true;
    }

    std::string FIFOSafeQueue::dump() const
    {
        std::string ret;
        for (const auto& iter : bufferList_) {
            std::stringstream ss;
            ss << "index:" << iter.first << " " << iter.second->capacity() << "  ";
            ret += ss.str();
        }
        return ret;
    }

    void FIFOSafeQueue::updateWritePtr()
    {
        // 写指针的改变 以及使用都是在写线程操作，因此不存在多线程竞争问题
        std::lock_guard<std::mutex> lk(mutex_);
        int                         maxIndex = bufferList_.rbegin()->first;
        queue_size_t                oldCap   = bufferList_.rbegin()->second->capacity();
        queue_size_t                newcap   = oldCap * 2;
        if (oldCap == types::queue_size_t_limit<queue_size_t>::max()) {
            newcap = oldCap;
        }
        maxIndex++;
        auto ptr              = std::make_shared<FIFOSafeElement>(newcap);  // 扩容
        bufferList_[maxIndex] = ptr;
        writeptr_             = ptr;
    }

    void FIFOSafeQueue::updateReadPtr()
    {
        // 读指针的改变 以及使用都是在读线程操作，因此不存在多线程竞争问题
        std::lock_guard<std::mutex> lk(mutex_);
        int                         originIndex = lastReadIndex_;
        lastReadIndex_++;
        auto iter = bufferList_.find(lastReadIndex_);
        if (iter != bufferList_.end()) {
            readptr_ = iter->second;
            bufferList_.erase(originIndex);
        }
        else {
            TRACE_ERROR << "what happened " << bufferList_.size() << " " << originIndex << " " << lastReadIndex_
                        << " buffer cap " << readptr_->capacity();
            lastReadIndex_--;
        }
    }

    bool FIFOSafeQueue::tryWrite(const char* buffer, queue_size_t len) { return writeptr_->write(buffer, len); }

    void FIFOSafeQueue::write(const char* buffer, queue_size_t len)
    {
        bool ret = writeptr_->write(buffer, len);
        if (!ret) {  // buffer full failed
            if (autoExpand_) {
                updateWritePtr();               // expand
                writeptr_->write(buffer, len);  // 只扩容一次
            }
        }
    }
    void FIFOSafeQueue::writeOver(queue_size_t len)
    {
        writeptr_->writeOver(len);
        if (isNeedNotify_) {
            uint64_t one = 1;
            auto     n   = notifyfd_->write(one);
            if (n != sizeof one) {
                SYSTEM_ERROR << "event write failed fd:"_s << notifyfd_->Fd() << " queue id:"_s << queueId_;
            }
        }
    }

    void FIFOSafeQueue::writeOverNotNotify(queue_size_t len) { writeptr_->writeOver(len); }

    char* FIFOSafeQueue::read(queue_size_t len)
    {
        char* buf = readptr_->read(len);
        if (autoExpand_) {
            if (buf == nullptr) {
                bool needUpdate = true;
                if (!isNeedNotify_) {  // 不需要通知就进行读操作的read
                    // 可能没有数据的时候就开始读了，这个时候自动扩容的情况下 没有扩容情况下 不更新读指针
                    std::size_t listSize;
                    {
                        std::lock_guard<std::mutex> lk(mutex_);
                        listSize = bufferList_.size();
                    }
                    if (listSize == 1) {
                        needUpdate = false;
                    }
                }
                if (needUpdate) {
                    buf = readptr_->read(
                        len);  // 第一次读的时候发现是空，正准备切换buffer，恰好又有数据写入了 所以还不能切换
                    if (buf) {
                        return buf;
                    }
                    else {
                        updateReadPtr();
                        buf = readptr_->read(len);
                    }
                }
            }
        }
        return buf;
    }
    void FIFOSafeQueue::readOver(types::queue_size_t len) { readptr_->readOver(len); }
}  // namespace msg
}  // namespace xncc